package streating

import org.apache.spark.SparkConf
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object s5 {
  def main(args: Array[String]): Unit = {
    // 设置Windows环境下Hadoop home路径（解决可能出现的winutils错误）
    System.setProperty("hadoop.home.dir", "C:\\path\\to\\hadoop") // 替换为你的 hadoop路径
    // 创建Spark配置
    val conf = new SparkConf()
      .setAppName("WindowsSocketCSVProcessor")
      .setMaster("local[2]") // 使用2个本地线程
    // 创建StreamingContext，批次间隔5秒
     val ssc = new StreamingContext(conf, Seconds(5))
    // 从socket源创建DStream，监听本地9999端口
    val lines = ssc.socketTextStream("localhost", 9999)
    // 定义CSV数据的Schema（根据实际CSV结构调整）
    val schema = StructType(Array(
      StructField("id", IntegerType, nullable = true),
      StructField("name", StringType, nullable = true),
      StructField("age", IntegerType, nullable = true),
      StructField("city", StringType, nullable = true)   ))
    // 处理CSV数据
    val processedData = lines.flatMap { line =>
      try { // 简单的CSV解析（假设使用逗号分隔）
         val fields = line.split(",").map(_.trim)
        if (fields.length == schema.length) {
          Some(Row(fields(0).toInt,
          // id
          fields(1),
          // name
          fields(2).toInt,
          // age
         fields(3)
          // city
          ))
        } else {
          None
        }
      } catch {
        case e: Exception =>println(s"Error parsing line: $line, ${e.getMessage}")
        None
      }
    }
     // 统计操作1：按城市统计人数
    processedData.foreachRDD { rdd =>
      if (!rdd.isEmpty()) { // 创建SparkSession
         val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate()
    import spark.implicits._
         // 创建DataFrame
        val df = spark.createDataFrame(rdd, schema)
        // 按城市统计
        val cityCounts = df.groupBy("city").count()
        println("=== 按城市统计人数 ===")
        cityCounts.show()
        // 年龄统计
        val ageStats = df.describe("age")
        println("=== 年龄统计 ===")
        ageStats.show()
        // 可以添加更多统计或保存到文件/数据库
        }   }
         // 统计操作2：实时打印接收到的记录数
       lines.count().map(cnt => s"Received $cnt records").print()
         // 启动流计算
       ssc.start()
       ssc.awaitTermination()
        }
  }
